package com.example.day04.controller.sideoutputs;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

public class WaterDemo4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
//        env.setParallelism(1);
        List<WaterEntityDemo> list = Arrays.asList(
                new WaterEntityDemo("s1", 1L, 1),
                new WaterEntityDemo("s2", 1L, 1),
                new WaterEntityDemo("s3", 2L, 1),
                new WaterEntityDemo("s3", 1L, 1), // 注意：id 为空，因为第二个元素缺少 id
                new WaterEntityDemo("s2", 1L, 1),
                new WaterEntityDemo("s2", 12L, 1),
                new WaterEntityDemo("s2", 13L, 1),
                new WaterEntityDemo("s2", 14L, 1),
                new WaterEntityDemo("s1", 1L, 1),
                new WaterEntityDemo("s1", 2L, 1),
                new WaterEntityDemo("s1", 3L, 1),
                new WaterEntityDemo("s1", 5L, 1),
                new WaterEntityDemo("s1", 9L, 1),
                new WaterEntityDemo("s1", 10L, 1),
                new WaterEntityDemo("s1", 12L, 2),
                new WaterEntityDemo("s2", 12L, 3),
                new WaterEntityDemo("s1", 16L, 22),
                new WaterEntityDemo("s2", 16L, 33),
                new WaterEntityDemo("s3", 14L, 1),
                new WaterEntityDemo("s2", 16L, 33),
                new WaterEntityDemo("s3", 16L, 2),
                new WaterEntityDemo("s3", 20L, 1)
        );

        WaterEntityDemo[] array = list.toArray(new WaterEntityDemo[0]);

        SingleOutputStreamOperator<WaterEntityDemo> waterEntitySingleOutputStreamOperator = env.fromElements(array).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterEntityDemo>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((element, recordTimestamp) -> element.getVc() * 1000L));


        SingleOutputStreamOperator<Tuple3<String, Integer, Long>> aggregate = waterEntitySingleOutputStreamOperator.keyBy(WaterEntityDemo::getId).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .aggregate(new MyAggregateFunction(), new MyWindowFunction());
        aggregate.keyBy(r -> r.f2).process(new TopNn(2)).print();

//        Top-1: id=s1, count=4窗口的开始时间结束时间：-5000 - 5000
//        Top-2: id=s3, count=2窗口的开始时间结束时间：-5000 - 5000
//        Top-3: id=s2, count=2窗口的开始时间结束时间：-5000 - 5000
//        Top-1: id=s1, count=6窗口的开始时间结束时间：0 - 10000
//        Top-2: id=s3, count=2窗口的开始时间结束时间：0 - 10000
//        Top-3: id=s2, count=2窗口的开始时间结束时间：0 - 10000
//        Top-1: id=s1, count=4窗口的开始时间结束时间：5000 - 15000
//        Top-2: id=s2, count=4窗口的开始时间结束时间：5000 - 15000
//        Top-3: id=s3, count=1窗口的开始时间结束时间：5000 - 15000

        env.execute("aaa");
    }

    public static class MyAggregateFunction implements AggregateFunction<WaterEntityDemo, Integer, Integer> {

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(WaterEntityDemo value, Integer accumulator) {
            return accumulator + 1;
        }

        @Override
        public Integer getResult(Integer accumulator) {
            return accumulator;
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return null;
        }
    }

    public static class MyWindowFunction extends ProcessWindowFunction<Integer, Tuple3<String, Integer, Long>, String, TimeWindow> {


        @Override
        public void process(String key, ProcessWindowFunction<Integer, Tuple3<String, Integer, Long>, String, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Tuple3<String, Integer, Long>> out) throws Exception {

            Integer next = elements.iterator().next();
            out.collect(Tuple3.of(key, next, context.window().getEnd()));

        }
    }

    public static class TopNn extends KeyedProcessFunction<Long, Tuple3<String, Integer, Long>, String> {

        private Map<Long, List<Tuple3<String, Integer, Long>>> dataMaps;

        private Integer topN;
        public TopNn(Integer topN) {
            this.topN = topN;
            this.dataMaps = new HashMap<>();
        }

        @Override
        public void processElement(Tuple3<String, Integer, Long> value, KeyedProcessFunction<Long, Tuple3<String, Integer, Long>, String>.Context ctx, Collector<String> out) throws Exception {
            if (dataMaps.containsKey(value.f2)) {
                dataMaps.get(value.f2).add(value);
            } else {
                List<Tuple3<String, Integer, Long>> list = new ArrayList<>();
                list.add(value);
                dataMaps.put(value.f2, list);
            }
            //
            ctx.timerService().registerEventTimeTimer(value.f2 + 1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            Long currentKey = ctx.getCurrentKey();
            List<Tuple3<String, Integer, Long>> list = dataMaps.get(currentKey);
            list.sort((o1, o2) -> o2.f1.compareTo(o1.f1)); // 使用 compareTo 而不是减法
            // 输出 Top 3 结果
//            int topN = 2;
            for (int i = 0; i < Math.min(list.size(), this.topN); i++) {
                Tuple3<String, Integer, Long> data = list.get(i);
                out.collect("Top-" + (i + 1) + ": id=" + data.f0 + ", count=" + data.f1 + "窗口的开始时间结束时间：" +data.f2);
            }

            list.clear();
        }
    }
    public static class MyTopNpAWF extends ProcessAllWindowFunction<WaterEntityDemo, String, TimeWindow> {
        @Override
        public void process(Context context, Iterable<WaterEntityDemo> elements, Collector<String> out) throws Exception {
            Map<String, Integer> map = new HashMap<>();

            for (WaterEntityDemo element : elements) {
                String id = element.getId();
                if (!map.containsKey(id)) {
                    map.put(id, 1);
                } else {
                    map.put(id, map.get(id) + 1);
                }
            }

            List<Tuple2<String, Integer>> datas = new ArrayList<>();
            for (String key : map.keySet()) {
                datas.add(Tuple2.of(key, map.get(key)));
            }

            // 降序排列
            datas.sort((o1, o2) -> o2.f1.compareTo(o1.f1)); // 使用 compareTo 而不是减法

            // 输出 Top 3 结果
            int topN = 3;
            for (int i = 0; i < Math.min(datas.size(), topN); i++) {
                Tuple2<String, Integer> data = datas.get(i);
                out.collect("Top-" + (i + 1) + ": id=" + data.f0 + ", count=" + data.f1 + "窗口的开始时间结束时间：" + context.window().getStart() + " - " + context.window().getEnd());
            }
        }
    }
}
